package com.yox.rpc.transport.netty.client;

import com.yox.rpc.codec.CommonDecoder;
import com.yox.rpc.codec.CommonEncoder;
import com.yox.rpc.serializer.CommonSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;


/**
 * 将Netty客户端建立Channel连接分离出来。
 *
 * 设定最大重连次数
 */
public class ChannelProvider {

    private static final Logger logger = LoggerFactory.getLogger(ChannelProvider.class);

    private static EventLoopGroup eventLoopGroup;
    private static Bootstrap bootstrap = initializeBootstrap();

    /**
     * 所有客户端Channel都保存在该Map中
     * key ip+序列器编号
     * value channel
     */
    private static Map<String, Channel> channels = new ConcurrentHashMap<>();

    private static Bootstrap initializeBootstrap() {
        eventLoopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                // 连接的超时时间，超时代表连接失败
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                // 是否开启 TCP 底层心跳机制
                .option(ChannelOption.SO_KEEPALIVE, true)
                // TCP 默认开启类 Nagle 算法，该算法的作用是尽可能的发送大数据块，减少网络传输。这里设置true手动开启 Nagle
                .option(ChannelOption.TCP_NODELAY, true);
        return bootstrap;
    }

    public static Channel getChannel(InetSocketAddress ip, CommonSerializer serializer) {
        // 先去Map里面读取channel
        String key = ip.toString() + serializer.getCode();
        if (channels.containsKey(key)) {
            Channel channel = channels.get(key);
            if (channel != null && channel.isActive()) {
                return channel;
            } else {
                channels.remove(key);
            }
        }

        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                // 设定IdleStateHandler心跳检测每5秒进行一次写检测，如果5秒内write()方法未被调用则触发一次userEventTrigger()方法
                // 实现客户端每5秒向服务端发送一次消息
                pipeline.addLast(new CommonEncoder(serializer))
                        .addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS))
                        .addLast(new CommonDecoder())
                        .addLast(new NettyClientHandler());
            }
        });

        Channel channel = null;
        try {
            channel = connect(bootstrap, ip);
        } catch (ExecutionException | InterruptedException e) {
            logger.error("连接客户端时发生错误", e);
            return null;
        }
        channels.put(key, channel);
        return channel;
    }

    /**
     *  Netty 客户端创建通道连接
     * @param bootstrap
     * @param ip
     */
    private static Channel connect(Bootstrap bootstrap, InetSocketAddress ip) throws ExecutionException, InterruptedException {
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        bootstrap.connect(ip).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                logger.info("客户端连接成功！");
                completableFuture.complete(future.channel());
            }else {
                throw new IllegalStateException();
            }
        });
        return completableFuture.get();
    }
}
